package com.xdf.testkafk.kafka;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;

import java.util.Optional;
import java.util.concurrent.TimeUnit;

/**
 * @author chanchaw
 * @create 2025-10-31 12:17
 */
@RequiredArgsConstructor
@Component
@Slf4j
public class KafkaProducer {
    private final KafkaTemplate<String,String> kafkaTemplate;

    /**
     * 发送 kafka 消息
     */
    public void sendMessage(final String key, String value, final String topic){
        //#region 检查传入参数是否合法
        if(null == value || null == topic) throw new IllegalArgumentException("发送kafka消时要求 value 和 topic 都不为空！");
        if(value.length() == 0 || topic.length() == 0) throw new IllegalArgumentException("发送kafka消时要求 value 和 topic 都不为空！");
        //#endregion

        ListenableFuture<SendResult<String,String>> future = StringUtils.isBlank(key) ? kafkaTemplate.send(topic,value) : kafkaTemplate.send(topic,key,value);
        // 异步回调的方式获取通知
        future.addCallback(
                success -> {
                    assert null != success && null != success.getRecordMetadata();
                    String _topic = success.getRecordMetadata().topic();
                    int _partition = success.getRecordMetadata().partition();
                    long _offset = success.getRecordMetadata().offset();

                    log.info("向 kafka 发送消息成功，topic:[{}],partition:[{}],offset:[{}]", _topic, _partition, _offset);
                }, failure -> {
                    log.error("向 kafka 发送消息失败，topic:[{}],key:[{}],value:[{}], 失败提示消息：[{}]", topic,key,value,failure.getMessage(), failure);
                }
        );

        // 同步的方式获取通知
        try {
            SendResult<String,String> sendResult = future.get(5, TimeUnit.SECONDS);// 5秒还没有获取到数据视作超时
        } catch (Exception ex){
            log.error("向 kafka 发送消息失败，topic:[{}],key:[{}],value:[{}], 失败提示消息：[{}]", topic,key,value,ex.getMessage(), ex);
        }

    }
}
